package demo.mq.rabbit;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import demo.algorithm.Fibonacci;
import demo.java.lang.ThreadDemo;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeoutException;

public class ConsumerDemo {

    private static Logger logger = LoggerFactory.getLogger(ConsumerDemo.class);

    private final static String QUEUE_NAME = "q_test_01";

    /**
     * 消费限流，并自动acknowledgement
     */
    @Test
    public void receiveMsgQosAutoAck() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // accept only one unack-ed message at a time (see below)
            channel.basicQos(1);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");

                System.out.println(" [x] Received '" + message + "'");
                try {
                    ThreadDemo.safeSleep(1000);
                } finally {
                    System.out.println(" [x] Done");
                }
            };
            // acknowledgment is covered below
            boolean autoAck = true;
            channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, new CancelCallback() {
                @Override
                public void handle(String consumerTag) throws IOException {
                    logger.info("consumerTag : {} ", consumerTag);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }

    @Test
    public void receiveMsgByDirectExchange() {
        String exchangeName = "direct_logs";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(exchangeName, "direct");
            /*
             * queueDeclare() create a non-durable, exclusive, autoDelete queue with a
             * generated name:
             */
            String queueName = channel.queueDeclare().getQueue();
            String routingKey = "0";
            // 绑定 queue 和 exchange
            channel.queueBind(queueName, exchangeName, routingKey);

            // accept only one unack-ed message at a time (see below)
            channel.basicQos(1);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
            };
            // acknowledgment is covered below
            boolean autoAck = true;
            channel.basicConsume(queueName, autoAck, deliverCallback, new CancelCallback() {
                @Override
                public void handle(String consumerTag) throws IOException {
                    logger.info("CancelCallback , consumerTag : {} ", consumerTag);
                    ThreadDemo.safeSleep(500);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }

    public static final String EXCHANGE_NAME = "test.xdcs.exchange";

    @Test
    public void receiveMsgByDirectExchangeAndAck() throws IOException, TimeoutException {
        Random random = new Random();
        String exchangeName = EXCHANGE_NAME;
        try (Connection connection = RabbitMqDemo.getConnection();
             Channel channel = connection.createChannel();) {
            channel.exchangeDeclare(exchangeName, "direct");
            /*
             * queueDeclare() create a non-durable, exclusive, autoDelete queue with a
             * generated name:
             */
            String queueName = channel.queueDeclare().getQueue();
            String routingKey = "0";
            // 绑定 queue 和 exchange
            channel.queueBind(queueName, exchangeName, routingKey);

            // accept only one unack-ed message at a time (see below)
            channel.basicQos(1);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
                boolean ok = random.nextBoolean();
                if (ok) {
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    System.out.println("消费成功");
                } else {
                    System.out.println("消费失败");
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                }
            };
            // acknowledgment is covered below
            boolean autoAck = false;
            channel.basicConsume(queueName, autoAck, deliverCallback, new CancelCallback() {
                @Override
                public void handle(String consumerTag) throws IOException {
                    logger.info("CancelCallback , consumerTag : {} ", consumerTag);
                    ThreadDemo.safeSleep(500);
                }
            });
        }
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }

    @Test
    public void receiveMsgByTopicExchange() throws IOException, TimeoutException {
        String exchangeName = EXCHANGE_NAME;
        try (Connection connection = RabbitMqDemo.getConnection();
             Channel channel = connection.createChannel();) {
            channel.exchangeDeclare(exchangeName, "topic");
            /*
             * queueDeclare() create a non-durable, exclusive, autoDelete queue with a
             * generated name:
             */
            String queueName = channel.queueDeclare().getQueue();
            System.out.println("queue name: " + queueName);
            String routingKey = "*";
            // 绑定 queue 和 exchange
            channel.queueBind(queueName, exchangeName, routingKey);

            // accept only one unack-ed message at a time (see below)
            channel.basicQos(1);

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
            };
            // acknowledgment is covered below
            boolean autoAck = true;
            channel.basicConsume(queueName, autoAck, deliverCallback, new CancelCallback() {
                @Override
                public void handle(String consumerTag) throws IOException {
                    logger.info("consumerTag : {} ", consumerTag);
                    ThreadDemo.safeSleep(1000);
                }
            });
        }

        ThreadDemo.safeSleep(3 * 60 * 1000);
    }

    /**
     * <link>https://www.rabbitmq.com/tutorials/tutorial-six-java.html</link>
     */
    @Test
    public void rpcServer() {
        String RPC_QUEUE_NAME = "rpc_queue";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
            channel.queuePurge(RPC_QUEUE_NAME);

            channel.basicQos(1);

            System.out.println(" [x] Awaiting RPC requests");

            Object monitor = new Object();
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                        .correlationId(delivery.getProperties().getCorrelationId()).build();

                String response = "";

                try {
                    String message = new String(delivery.getBody(), "UTF-8");
                    int n = Integer.parseInt(message);

                    System.out.println(" [.] fib(" + message + ")");
                    response += Fibonacci.fibonacciA(n);
                } catch (RuntimeException e) {
                    System.out.println(" [.] " + e.toString());
                } finally {
                    channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps,
                            response.getBytes("UTF-8"));
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    // RabbitMq consumer worker thread notifies the RPC server owner thread
                    synchronized (monitor) {
                        monitor.notify();
                    }
                }
            };

            channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> {
            }));
            // Wait and be prepared to consume the message from RPC client.
            while (true) {
                synchronized (monitor) {
                    try {
                        monitor.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}
